#include "EventEngine/Excutor.h"

#define EXCUTORBUFF 512

namespace NTrader
{

Excutor::Excutor()
	: __started(false)
	, __waitStrategy()
	, __consumed(__waitStrategy, EXCUTORBUFF - 1)
	, __buffer(EXCUTORBUFF)
	, __claimStrategy(EXCUTORBUFF, __waitStrategy)
	, __status(WAITING)
	, __nextToRead(0)
{
	__excutor.clear();
	__claimStrategy.add_claim_barrier(__consumed);
}


Excutor::~Excutor()
{
}

void Excutor::Excute()
{
	while (__started.load(std::memory_order_acquire))
	{
		disruptorplus::sequence_t toRead = __nextToRead.fetch_add(1, std::memory_order_relaxed);
		const auto available = __claimStrategy.wait_until_published(toRead,std::chrono::microseconds(10));
		if (available < toRead)
		{
			continue;
		}

		Actor_t a = __buffer[toRead];
		__consumed.publish(toRead + EXCUTORBUFF);

		a->Excute();
	}
}

void Excutor::Post(Actor_t a)
{
	const auto seq = __claimStrategy.claim_one();
	__buffer[seq] = a;
	__claimStrategy.publish(seq);
}

void Excutor::Start()
{
	if(!__started.load(std::memory_order_acquire))
	{
		__started.store(true, std::memory_order_release);
		
		for (size_t i = 0; i < std::thread::hardware_concurrency() * 2; i++)
		{
			__excutor.push_back(new std::thread(std::bind(&Excutor::Excute, this)));
		}
	}	
}

void Excutor::Stop()
{
	if (__started.load(std::memory_order_acquire))
	{
		__started.store(false, std::memory_order_release);

		for (auto item : __excutor)
		{
			item->detach();
			delete item;
		}
		
		__excutor.clear();
	}
}

}
